package com.zyx.flinkdemo.stream.state.keyed;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author Yaxi.Zhang
 * @since 2021/6/2 15:58
 * desc: TTLState案例
 *  https://ci.apache.org/projects/flink/flink-docs-release-1.13/zh/docs/dev/datastream/fault-tolerance/state/
 *
 *  StateTtlConfig ttlConfig = StateTtlConfig
 *     .newBuilder(Time.seconds(1))
 *     .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
 *     .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
 *     .build();
 *  TTL 配置有以下几个选项：
 *      newBuilder 的第一个参数表示数据的有效期，是必选项。
 *      TTL 的更新策略（默认是 OnCreateAndWrite）：
 *          StateTtlConfig.UpdateType.OnCreateAndWrite - 仅在创建和写入时更新
 *          StateTtlConfig.UpdateType.OnReadAndWrite - 读取时也更新
 *      数据在过期但还未被清理时的可见性配置如下（默认为 NeverReturnExpired):
 *          StateTtlConfig.StateVisibility.NeverReturnExpired - 不返回过期数据
 *          StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp - 会返回过期但未清理的数据
 *          NeverReturnExpired 情况下，过期数据就像不存在一样，不管是否被物理删除。这对于不能访问过期数据的场景下非常有用，比如敏感数据。
 *          ReturnExpiredIfNotCleanedUp 在数据被物理删除前都会返回。
 */
public class TtlStateFlatMap extends RichFlatMapFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> {
    public static Logger log = LoggerFactory.getLogger(TtlStateFlatMap.class);
    public transient ValueState<Tuple2<Integer, Integer>> valueState;
    private final boolean isRead;
    private final boolean isReturn;

    public TtlStateFlatMap(boolean isRead, boolean isReturn) {
        this.isRead = isRead;
        this.isReturn = isReturn;
    }

    @Override
    public void open(Configuration config) {
        log.info("{},{}", Thread.currentThread().getName(), "恢复或初始化状态");
        ValueStateDescriptor<Tuple2<Integer, Integer>> descriptor = new ValueStateDescriptor<>(
                "ValueStateTTL",
                TypeInformation.of(new TypeHint<Tuple2<Integer, Integer>>() {
                }));
        StateTtlConfig.Builder builder = StateTtlConfig.newBuilder(Time.seconds(6));
        if (isRead) {
            builder = builder.setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite);
        } else {
            builder = builder.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite);
        }
        if (isReturn) {
            builder = builder.setStateVisibility(StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp);
        } else {
            builder = builder.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired);
        }
        StateTtlConfig ttlConfig = builder.build();
        descriptor.enableTimeToLive(ttlConfig);
        valueState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void flatMap(Tuple2<Integer, Integer> input, Collector<Tuple2<Integer, Integer>> out) throws Exception {
        if (input.f1 > 10 && input.f1 < 20) {
            if (isRead) {
                log.info("不执行操作:{}", input);
                out.collect(input);
            } else {
                Tuple2<Integer, Integer> currentSum = valueState.value();
                log.info("不执行操作 状态值:{} 输入值:{}", currentSum, input);
            }
        } else {
            Tuple2<Integer, Integer> currentSum = valueState.value();
            log.info("currentSum before:" + currentSum);
            System.out.println("currentSum before:" + currentSum);
            if (currentSum == null) {
                currentSum = input;
            } else {
                currentSum.f1 = currentSum.f1 + input.f1;
            }
            valueState.update(currentSum);
            log.info("currentSum after:" + currentSum);
            out.collect(currentSum);
        }
    }

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        KeyedStream<Tuple2<Integer, Integer>, Tuple> keyedStream = KeyStateBase.before(env);
        keyedStream.flatMap(new TtlStateFlatMap(true, false)).print("输出结果 >>> ");
        env.execute("TTLStateFlatMap >>> ");
    }
}
